package syan.rediss.config;

import syan.rediss.liststener.ListenerMessage;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * @author CHENTAIWEN
 * @create 2022/8/4 9:39
 */
@Slf4j
@Configuration
public class RedisStreamConfig {

    @Autowired
    private ListenerMessage streamListener;
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Value("${redis.Stream}")
    private String stream;
    @Value("${redis.Group}")
    private String groupId;
    @Value("${redis.name}")
    private String name;

    @Bean
    public Subscription subscription(RedisConnectionFactory factory){
        var options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(5))//超时配置
                .batchSize(1)//单次消费最多count
                .build();
        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
        initStream(stream, groupId);
        //监听的是消息组接受到的消息，而不是消费过的消息
        var subscription = listenerContainer.receive(Consumer.from(groupId, name),
                StreamOffset.create(stream, ReadOffset.lastConsumed()), streamListener);
        listenerContainer.start();
        return subscription;
    }
/*    @Bean
    public Subscription subscription2(RedisConnectionFactory factory) {

        synchronized (factory) {
            String stream = key + "demo";
            String groupId = "demo";
            String name = "yingzio";
            var options = StreamMessageListenerContainer
                    .StreamMessageListenerContainerOptions
                    .builder()
                    .batchSize(1)
                    .build();
            var listenerContainer = StreamMessageListenerContainer.create(factory, options);
            initStream(stream, groupId);
            var subscription = listenerContainer.receiveAutoAck(Consumer.from(groupId, name),
                    StreamOffset.create(stream, ReadOffset.lastConsumed()), streamListener);
            listenerContainer.start();
            return subscription;
        }
    }
    @Bean
    public Subscription subscription3(RedisConnectionFactory factory) {
        synchronized (factory) {
            String stream = key + "demo";
            String groupId = "demo";
            String name = "yingzio";
            var options = StreamMessageListenerContainer
                    .StreamMessageListenerContainerOptions
                    .builder()
                    .batchSize(1)
                    .build();
            var listenerContainer = StreamMessageListenerContainer.create(factory, options);
            initStream(stream, groupId);
            var subscription = listenerContainer.receiveAutoAck(Consumer.from(groupId, name),
                    StreamOffset.create(stream, ReadOffset.lastConsumed()), streamListener);
            listenerContainer.start();
            return subscription;
        }
    }
    @Bean
    public Subscription subscription4(RedisConnectionFactory factory) {
        synchronized (factory) {
            String stream = key + "demo";
            String groupId = "demo";
            String name = "yingzio";
            var options = StreamMessageListenerContainer
                    .StreamMessageListenerContainerOptions
                    .builder()
                    .batchSize(1)
                    .build();
            var listenerContainer = StreamMessageListenerContainer.create(factory, options);
            initStream(stream, groupId);
            var subscription = listenerContainer.receiveAutoAck(Consumer.from(groupId, name),
                    StreamOffset.create(stream, ReadOffset.lastConsumed()), streamListener);
            listenerContainer.start();
            return subscription;
        }
    }
    @Bean
    public Subscription subscription5(RedisConnectionFactory factory) {
        synchronized (factory) {
            String stream = key + "demo";
            String groupId = "demo";
            String name = "yingzio";
            var options = StreamMessageListenerContainer
                    .StreamMessageListenerContainerOptions
                    .builder()
                    .batchSize(1)
                    .build();
            var listenerContainer = StreamMessageListenerContainer.create(factory, options);
            initStream(stream, groupId);
            var subscription = listenerContainer.receiveAutoAck(Consumer.from(groupId, name),
                    StreamOffset.create(stream, ReadOffset.lastConsumed()), streamListener);
            listenerContainer.start();
            return subscription;
        }
    }*/
    private void initStream(String key, String group){
        //判断key是否存在，如果不存在则创建
        boolean hasKey=false;
        if (key != null){
            hasKey = redisTemplate.hasKey(key);
        }
        if(!hasKey){
            Map<String,Object> map = new HashMap<>();
            map.put("field","value");
            RecordId recordId = redisTemplate.opsForStream().add(key, map);
            redisTemplate.opsForStream().createGroup(key, ReadOffset.from("0-0"), group);
            redisTemplate.opsForStream().delete(key,recordId.getValue());
            log.error("stream:{}-group:{} initialize success",key,group);
        }else{
            try {
                redisTemplate.opsForStream().createGroup(key, ReadOffset.from("0-0"), group);
            }catch (Exception e){
//                如果消费组存在，删除再创建，用于读取全部信息
//                redisTemplate.opsForStream().destroyGroup(key, group);
//                redisTemplate.opsForStream().createGroup(key, ReadOffset.from("0-0"), group);
            }
        }
    }


}